package com.lvlian

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema
import com.ververica.cdc.connectors.mysql.table.StartupOptions
import com.ververica.cdc.connectors.mysql.source.MySqlSource
import io.debezium.data.Json
import io.debezium.data.Envelope
import io.debezium.data.Envelope.Operation
import org.apache.kafka.connect.source.SourceRecord
import org.apache.kafka.connect.data.Schema
import org.apache.kafka.connect.data.Struct
import org.apache.kafka.connect.source.SourceTaskContext

import java.util
import java.util.Properties
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import com.ververica.cdc.connectors.mysql.source.MySqlSource
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema

import java.util.Properties
import com.alibaba.fastjson.JSON
import com.utils.OSUtils
import org.apache.flink.streaming.api.scala

/*
*SHOW VARIABLES LIKE 'binlog_format'; 设置为行级别
*
* /path/to/flink/bin/flink run /home/user/app/lvlian-cdc.jar /path/to/config.yaml
mysql -h192.168.0.26 -u root -pgni2PbgxMNutdxz
*  flink run -c com.lvlian.LvlianCdc2  cq-flink-cdc-1.0-SNAPSHOT.jar product.yaml
* */

//java -jar target/your-jar-file.jar /path/to/config.yaml

object LvlianCdc2 {
  def main(args: Array[String]): Unit = {


    var startPos = StartupOptions.latest()
    var fileName = "product.yaml"

    if (args.length > 0 && args(0).equals("true")) {

      startPos = StartupOptions.initial()
    }
    print(startPos + args(0) + "   输出数据库信息  =====" + fileName)

    if (OSUtils.isWindows) {
      fileName = "dev.yaml"
    }

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val scf_recycling_in_infoTag = new OutputTag[String]("scf_recycling_in_info")
    val scf_renewable_in_infoTag = new OutputTag[String]("scf_renewable_in_info")
    val scf_invoice_infoTag = new OutputTag[String]("scf_invoice_info")
    val scf_contract_infoTag = new OutputTag[String]("scf_contract_info")
    val scf_payment_infoTag = new OutputTag[String]("scf_payment_info")
    val scf_subject_infoTag = new OutputTag[String]("scf_subject_basic_info")
    val scf_driver_infoTag = new OutputTag[String]("scf_driver_info")
    val scf_truck_infoTag = new OutputTag[String]("scf_truck_info")

    val source: scala.DataStream[String] = env.fromSource(
      LvlianSouce.createCDCSource(fileName, startPos),
      WatermarkStrategy.noWatermarks(),
      "mysql cdc source"
    )
    source.print.setParallelism(1)
    val processedStream = source.process(new ProcessFunction[String, String] {
      override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
        val json = JSON.parseObject(value)
        val tableName = json.getJSONObject("source").getString("table")
        tableName match {
          case "scf_recycling_in_info" => ctx.output(scf_recycling_in_infoTag, value)
          case "scf_renewable_in_info" => ctx.output(scf_renewable_in_infoTag, value)
          case "scf_invoice_info" => ctx.output(scf_invoice_infoTag, value)
          case "scf_contract_info" => ctx.output(scf_contract_infoTag, value)
          case "scf_payment_info" => ctx.output(scf_payment_infoTag, value)
          case "scf_driver_info" => ctx.output(scf_driver_infoTag, value)
          case "scf_truck_info" => ctx.output(scf_truck_infoTag, value)
          case "scf_subject_basic_info" => ctx.output(scf_subject_infoTag, value)
          case _ => out.collect(value)
        }
      }
    })
    processedStream.getSideOutput(scf_payment_infoTag).addSink(new LvlianSinkPaymentInfo).name("LvlianSinkPaymentInfoSink")
    processedStream.getSideOutput(scf_recycling_in_infoTag).addSink(new LvlianSinkRecyclingIn).name("LvlianSinkRecyclingIn Sink")
    processedStream.getSideOutput(scf_renewable_in_infoTag).addSink(new LvlianSinkRenewableIn).name("LvlianSinkRenewableIn  Sink")
    processedStream.getSideOutput(scf_invoice_infoTag).addSink(new LvlianSinkInvoice).name("LvlianSinkInvoice In Sink")
    processedStream.getSideOutput(scf_contract_infoTag).addSink(new LvlianSinkContractInfo).name("LvlianSinkContractInfo   Sink")
    processedStream.getSideOutput(scf_subject_infoTag).addSink(new LvlianSinkSubjectInfo).name("LvlianSinkSubjectInfoSink")
    processedStream.getSideOutput(scf_driver_infoTag).addSink(new LvlianSinkDriver).name("LvlianSinkDriver")
    processedStream.getSideOutput(scf_truck_infoTag).addSink(new LvlianSinkVehicle).name("LvlianSinkVehicle")


    val dtf = DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss")
    val now = LocalDateTime.now.format(dtf)

    env.execute("Flink Job at " + now)

  }
}